package com.shujia.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo5SourceFunction {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义source
    val myDS: DataStream[Int] = env.addSource(new MySource)

    myDS.print()

    env.execute()

  }

}

/**
  * 自定义source,实现SourceFunction接口
  *
  */
class MySource extends SourceFunction[Int] {
  /**
    * 读取数据的方法,只执行一次
    *
    * @param ctx : 用于将数据发生到下游
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
    /*val list: Range.Inclusive = 0 to 100

    for (i <- list) {
      //将数据发生到下游
      ctx.collect(i)
    }*/


    var i = 0
    //如果是是循环,得到的流是一个无界流
    while (true) {
      //将数据发生到下游
      ctx.collect(i)
      i += 1
    }
  }

  override def cancel(): Unit = {

  }
}
